package com.example.demo.kafka;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;

import java.util.ArrayList;
import java.util.Collection;

public class SocketSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        ArrayList<Tuple2<Integer, Integer>> list = new ArrayList<>();
        Tuple2<Integer, Integer> st = new Tuple2<>();
        st.f0 = 1;
        st.f1 = 2;
        list.add(st);
        DataStreamSource<Tuple2<Integer, Integer>> cc = env.fromCollection(list);

//        DataStream<String> text = env.readTextFile("/Users/van/Downloads/rsa/java/encrypt/HexUtil.java");


//        SingleOutputStreamOperator<Integer> map = text.map((MapFunction<String, Integer>) s -> {
//
//            if (StringUtils.isNotBlank(s)) {
//                return 1;
//            }
//            return 0;
//        });
//        map.print();
//        cc.map(new MapFunction<Tuple2<Integer, Integer>, Object>() {
//            @Override
//            public Object map(Tuple2<Integer, Integer> integerStringTuple2) throws Exception {
//                return null;
//            }
//        });
        cc.keyBy(0)
                .sum(1)
                .print();


        env.execute("console");
    }


}